"""
Case Type   : 发布订阅
Case Name   : 发布端为企业版本,订阅端为小型化版本,双向订阅内容相同
Create At   : 2023-01-30
Owner       : opentestcase004
Description :
    1.在两个集群创建表
    2.两个集群创建双向发布订阅
    3.集群A发布端插入数据
    4.集群B订阅端查询数据是否更新
    5.集群B发布端插入数据
    6.集群A订阅端查询数据是否更新
    7.修改集群B发布端
    8.集群A发布端插入数据
    9.集群B订阅端查询数据是否更新
    10.集群B发布端插入数据
    11.集群A订阅端查询数据是否更新
    12.清理环境
Expect      :
    1.成功
    2.成功
    3.成功
    4.表3数据不更新,其余数据更新
    5.成功
    6.表3数据不更新,其余数据更新
    7.成功
    8.成功
    9.表3数据不更新,其余数据更新
    10.成功
    11.表2数据更新,其余不更新
    12.成功
History     : 
"""

import os
import unittest
import subprocess
from yat.db.db import DB
from yat.shell import SSH
from yat.test.node.node import LazyResource
from yat.test import macro
from yat.test import Node
from testcase.utils.Common import Common
from testcase.utils.CommonSH import CommonSH
from testcase.utils.Constant import Constant
from testcase.utils.Logger import Logger
from testcase.utils.CommonLite import CommonLite

Primary_SH = CommonSH('PrimaryDbUser')


@unittest.skipIf(3 != Primary_SH.get_node_num(), '非1+2环境不执行')
class Pubsubclass(unittest.TestCase):
    def setUp(self):
        self.log = Logger()
        self.log.info("-----------this is setup-----------")
        self.log.info(f"-----{os.path.basename(__file__)} start-----")
        self.nodeA = Node('default')
        self.nodeB = None
        self.nodeB = self.nodeA
        self.nodes_yml = '/home/lite_user/port'
        self.log.info("-----------记录端口号-----------")
        self.file_exist = f'cat {self.nodes_yml}'
        self.log.info(self.file_exist)
        self.pri_root_nodes = Node(node='PrimaryRoot')
        self.port = self.pri_root_nodes.sh(self.file_exist).result()
        self.log.info(f"读取到的端口值是：{self.port}")
        self.assertIsInstance(int(self.port), int)
        self.nodeB.db = DB(f"{Node(node='PrimaryDbUser').db_name}", f"{macro.COMMON_PASSWD}",
                           f'{self.pri_root_nodes.ssh_host}', f'{self.port}')
        self.nodeB.ssh = LazyResource(
            lambda: SSH('lite_user', f"{macro.COMMON_PASSWD}", f'{self.pri_root_nodes.ssh_host}',
                        f"{Node(node='PrimaryDbUser').ssh_port}"))
        self.log.info(f' {self.nodeB.db.port}')
        self.log.info(f' {self.nodeB.name}')
        self.log.info(f' {self.nodeB.ssh_user}')
        self.log.info(f' {self.nodeB.ssh_password}')
        self.log.info(f' {self.nodeB.ssh_host}')
        self.log.info(f' {self.nodeB.ssh_port}')
        self.pri_userdb_pub = Node(node='PrimaryDbUser')
        self.pri_userdb_sub = self.nodeB
        self.constant = Constant()
        self.commsh_pub = CommonSH('PrimaryDbUser')
        self.commsh_sub = CommonLite(self.nodeB)
        self.com_sub = CommonLite(self.nodeB)
        self.com_pub = Common()
        self.db_instance_path_lite = '/home/lite_user/data'
        self.db_env_path_lite = '/home/lite_user/.bashrc'
        self.tb_name1 = 't_pub_sub_0314_01'
        self.tb_name2 = 't_pub_sub_0314_02'
        self.tb_name3 = 't_pub_sub_0314_03'
        self.subname = "s_pub_sub_0314"
        self.pubname = "p_pub_sub_0314"
        self.parent_path_pub = os.path.dirname(macro.DB_INSTANCE_PATH)
        self.parent_path_sub = os.path.dirname(self.db_instance_path_lite)
        self.port = str(int(self.pri_userdb_pub.db_port) + 1)
        self.port1 = str(int(self.pri_userdb_sub.db.port) + 1)
        self.wal_level = self.com_pub.show_param("wal_level")
        self.wal_level_sub = self.com_sub.show_param("wal_level",
                                                     self.db_env_path_lite)
        self.user_param_pub = f'-U {self.pri_userdb_pub.db_user} ' \
                              f'-W {self.pri_userdb_pub.db_password}'
        self.user_param_sub = f'-U {self.pri_userdb_sub.db_user} ' \
                              f'-W {self.pri_userdb_sub.db_password}'
        cmd = f"cp " \
              f"{os.path.join(macro.DB_INSTANCE_PATH, 'pg_hba.conf')} " \
              f"{os.path.join(self.parent_path_pub, 'pg_hba.conf')};"
        self.log.info(cmd)
        result = self.pri_userdb_pub.sh(cmd).result()
        self.log.info(result)
        cmd = f"cp " \
              f"{os.path.join(self.db_instance_path_lite, 'pg_hba.conf')}" \
              f" {os.path.join(self.parent_path_sub, 'pg_hba.conf')};"
        self.log.info(cmd)
        result = self.pri_userdb_sub.sh(cmd).result()
        self.log.info(result)
        self.hostname = self.pri_userdb_sub.sh('hostname').result().strip()
        text = "step:预置条件,修改pg_hba expect:成功"
        self.log.info(text)
        guc_res = self.commsh_pub.execute_gsguc(
            'reload', self.constant.GSGUC_SUCCESS_MSG, '',
            'all', False, False, '',
            f'host    replication  {self.pri_userdb_sub.db_user} '
            f'{self.pri_userdb_sub.db_host}/32 sha256')
        self.log.info(guc_res)
        self.assertTrue(guc_res, '执行失败' + text)
        guc_res = self.commsh_pub.execute_gsguc(
            'reload', self.constant.GSGUC_SUCCESS_MSG, '',
            'all', False, False, '',
            f'host    all  {self.pri_userdb_sub.db_user} '
            f'{self.pri_userdb_sub.db_host}/32 sha256')
        self.log.info(guc_res)
        self.assertTrue(guc_res, '执行失败' + text)
        result = self.commsh_pub.execute_gsguc(
            'set', self.constant.GSGUC_SUCCESS_MSG, 'wal_level=logical')
        self.assertTrue(result, '执行失败' + text)
        result = self.commsh_pub.restart_db_cluster(True)
        flg = self.constant.START_SUCCESS_MSG in result or 'Degrade' in result
        self.assertTrue(flg, '执行失败' + text)
        guc_res = self.commsh_sub.execute_gsguc(
            'reload', self.constant.GSGUC_SUCCESS_MSG, '',
            'all', False, True, self.db_instance_path_lite,
            f'host    replication  {self.pri_userdb_pub.db_user} '
            f'{self.pri_userdb_pub.db_host}/32 sha256',
            self.db_env_path_lite)
        self.log.info(guc_res)
        self.assertTrue(guc_res, '执行失败' + text)
        guc_res = self.commsh_sub.execute_gsguc(
            'reload', self.constant.GSGUC_SUCCESS_MSG, '',
            'all', False, True, self.db_instance_path_lite,
            f'host all  {self.pri_userdb_pub.db_user} '
            f'{self.pri_userdb_pub.db_host}/32 sha256',
            self.db_env_path_lite)
        self.log.info(guc_res)
        self.assertTrue(guc_res, '执行失败' + text)
        result = self.commsh_sub.execute_gsguc(
            'set', self.constant.GSGUC_SUCCESS_MSG, 'wal_level=logical',
            single=True, dn_path=self.db_instance_path_lite,
            pghba_param=False, env_path=self.db_env_path_lite)
        self.assertTrue(result, '执行失败' + text)
        result1 = self.commsh_sub.stop_db_instance(
            self.db_env_path_lite, self.db_instance_path_lite)
        self.assertTrue(result1, '执行失败' + text)
        result2 = self.commsh_sub.start_db_instance(
            env_path=self.db_env_path_lite,
            dn_path=self.db_instance_path_lite)
        self.assertTrue(result2, '执行失败' + text)

        result2 = self.commsh_sub.execute_gsguc(
            'set', self.constant.GSGUC_SUCCESS_MSG,
            "listen_addresses = '*'",
            single=True, dn_path=self.db_instance_path_lite,
            pghba_param=False, env_path=self.db_env_path_lite)
        result4 = self.commsh_sub.stop_db_instance(
            self.db_env_path_lite, self.db_instance_path_lite)
        result5 = self.commsh_sub.start_db_instance(
            env_path=self.db_env_path_lite,
            dn_path=self.db_instance_path_lite)
        self.log.info(result2)
        self.log.info(result4)
        self.log.info(result5)

    def test_pub_sub(self):
        text = "step1:在两个集群创建表 expect:成功"
        self.log.info(text)
        sql_cmd = f"drop table if exists {self.tb_name1},{self.tb_name2}," \
                  f"{self.tb_name3};" \
                  f"create table {self.tb_name1}(id number(7) constraint " \
                  f"s_longtext_id_nn not null, use_filename varchar2(20) primary " \
                  f"key, filename varchar2(255),text varchar2(2000));" \
                  f"create table {self.tb_name2}(like {self.tb_name1} including " \
                  f"all);" \
                  f"create table {self.tb_name3}(c1 int primary key, text " \
                  f"varchar(1024)) with (orientation = column, compression=high);"
        pub_result = self.commsh_pub.execut_db_sql(
            sql_cmd, sql_type=self.user_param_pub)
        self.log.info(pub_result)
        self.assertIn(self.constant.DROP_TABLE_SUCCESS, pub_result, '执行失败'
                      + text)
        self.assertEqual(pub_result.count(self.constant.TABLE_CREATE_SUCCESS),
                         6, '执行失败' + text)
        sub_result = self.commsh_sub.execut_db_sql(
            sql_cmd, self.user_param_sub, None, self.db_env_path_lite)
        self.log.info(sub_result)
        self.assertIn(self.constant.DROP_TABLE_SUCCESS, sub_result, '执行失败'
                      + text)
        self.assertEqual(sub_result.count(self.constant.TABLE_CREATE_SUCCESS),
                         6, '执行失败' + text)

        text = "step2:两个集群创建双向发布订阅 expect:成功"
        self.log.info(text)
        sql_cmd = f"drop publication if exists {self.pubname};" \
                  f"create publication {self.pubname} for all tables;"
        pub_result = self.commsh_pub.execut_db_sql(
            sql_cmd, sql_type=self.user_param_pub)
        self.log.info(pub_result)
        self.assertIn(self.constant.drop_pub_succ_msg, pub_result,
                      '执行失败' + text)
        self.assertIn(self.constant.create_pub_succ_msg, pub_result,
                      '执行失败' + text)
        result = self.commsh_sub.execute_generate(
            macro.COMMON_PASSWD, env_path=self.db_env_path_lite)
        self.assertIn(self.constant.create_keycipher_success, result,
                      '执行失败' + text)
        sql_cmd = f"drop subscription if exists {self.subname};" \
                  f"create subscription {self.subname} connection " \
                  f"'host={self.pri_userdb_pub.db_host} " \
                  f"port={self.port} user={self.pri_userdb_pub.db_user} " \
                  f"dbname={self.pri_userdb_pub.db_name} " \
                  f"password={self.pri_userdb_pub.db_password}' publication " \
                  f"{self.pubname};"
        sub_result = self.commsh_sub.execut_db_sql(
            sql_cmd, self.user_param_sub, None, self.db_env_path_lite)
        self.log.info(sub_result)
        self.assertIn(self.constant.drop_sub_succ_msg, sub_result,
                      '执行失败' + text)
        self.assertIn(self.constant.create_sub_succ_msg, sub_result, '执行失败'
                      + text)

        sql_cmd = f"drop publication if exists {self.pubname};" \
                  f"create publication {self.pubname} for table {self.tb_name1}," \
                  f"{self.tb_name2};"
        sub_result = self.commsh_sub.execut_db_sql(
            sql_cmd, self.user_param_sub, None, self.db_env_path_lite)
        self.log.info(pub_result)
        self.assertIn(self.constant.drop_pub_succ_msg, sub_result,
                      '执行失败' + text)
        self.assertIn(self.constant.create_pub_succ_msg, sub_result,
                      '执行失败' + text)
        result = self.commsh_pub.execute_generate(
            macro.COMMON_PASSWD)
        self.assertIn(self.constant.create_keycipher_success, result,
                      '执行失败' + text)
        sql_cmd = f"drop subscription if exists {self.subname};" \
                  f"create subscription {self.subname} connection " \
                  f"'host={self.pri_userdb_sub.db.host} " \
                  f"port={self.port1} user={self.pri_userdb_sub.db.user} " \
                  f"dbname={self.pri_userdb_sub.db_name} " \
                  f"password={self.pri_userdb_sub.db_password}' publication " \
                  f"{self.pubname};"
        pub_result = self.commsh_pub.execut_db_sql(
            sql_cmd, sql_type=self.user_param_pub)
        self.log.info(pub_result)
        self.assertIn(self.constant.drop_sub_succ_msg, pub_result,
                      '执行失败' + text)
        self.assertIn(self.constant.create_sub_succ_msg, pub_result, '执行失败'
                      + text)

        text = "step3:集群A发布端插入数据 expect:成功"
        self.log.info(text)
        sql_cmd = f"insert into {self.tb_name1} values(1, '1', " \
                  f"'tb_pubsub_case314_1', 'equal');" \
                  f"insert into {self.tb_name2} values(1, '1', " \
                  f"'tb_pubsub_case314_2', 'equal');" \
                  f"insert into {self.tb_name3} values(1, '1');"
        pub_result = self.commsh_pub.execut_db_sql(
            sql_cmd, sql_type=self.user_param_pub)
        self.log.info(pub_result)
        self.assertEqual(pub_result.count(self.constant.INSERT_SUCCESS_MSG), 3,
                         '执行失败' + text)

        text = "step4:集群B订阅端查询数据是否更新 expect:表3数据不更新," \
               "其余数据更新"
        self.log.info(text)
        sql_select = f"select pg_sleep(5.5);select * from {self.tb_name1};" \
                     f"select * from {self.tb_name2};" \
                     f"select * from {self.tb_name3};"
        sub_result = self.commsh_sub.execut_db_sql(
            sql_select, self.user_param_sub, None, self.db_env_path_lite)
        self.log.info(sub_result)
        self.assertIn('1 | 1            | tb_pubsub_case314_1 | equal',
                      sub_result, '执行失败' + text)
        self.assertIn('1 | 1            | tb_pubsub_case314_2 | equal',
                      sub_result, '执行失败' + text)
        self.assertEqual(str(sub_result.splitlines()[-1]).strip(), '(0 rows)',
                         '执行失败' + text)

        text = "step5:集群B发布端插入数据 expect:成功"
        self.log.info(text)
        sql_cmd = f"insert into {self.tb_name1} values(2, '2', " \
                  f"'tb_pubsub_case314_12', 'equal2');" \
                  f"insert into {self.tb_name2} values(2, '2', " \
                  f"'tb_pubsub_case314_22', 'equal2');" \
                  f"insert into {self.tb_name3} values(2, '2');"
        sub_result = self.commsh_sub.execut_db_sql(
            sql_cmd, self.user_param_sub, None, self.db_env_path_lite)
        self.log.info(sub_result)
        self.assertEqual(pub_result.count(self.constant.INSERT_SUCCESS_MSG), 3,
                         '执行失败' + text)

        text = "step6:集群A订阅端查询数据是否更新 expect:表3数据不更新," \
               "其余数据更新"
        self.log.info(text)
        sql_select = f"select pg_sleep(5.5);select * from {self.tb_name1};" \
                     f"select * from {self.tb_name2};" \
                     f"select * from {self.tb_name3};"
        pub_result = self.commsh_pub.execut_db_sql(
            sql_select, sql_type=self.user_param_pub)
        self.log.info(pub_result)
        self.assertIn('2 | 2            | tb_pubsub_case314_12 | equal2',
                      pub_result, '执行失败' + text)
        self.assertIn('2 | 2            | tb_pubsub_case314_22 | equal2',
                      pub_result, '执行失败' + text)
        self.assertEqual(str(pub_result.splitlines()[-2]).strip(), '1 | 1',
                         '执行失败' + text)
        self.assertEqual(str(pub_result.splitlines()[-1]).strip(), '(1 row)',
                         '执行失败' + text)

        text = "step7:修改集群B发布端 expect:成功"
        self.log.info(text)
        sql_cmd = f"alter publication {self.pubname} drop table " \
                  f"{self.tb_name1};"
        sub_result = self.commsh_sub.execut_db_sql(
            sql_cmd, self.user_param_sub, None, self.db_env_path_lite)
        self.log.info(sub_result)
        self.assertIn(self.constant.alter_pub_succ_msg, sub_result, '执行失败'
                      + text)

        text = "step8:集群A发布端插入数据 expect:成功"
        self.log.info(text)
        sql_cmd = f"insert into {self.tb_name1} values(3, '3', " \
                  f"'tb_pubsub_case314_1', 'equal3');" \
                  f"insert into {self.tb_name2} values(3, '3', " \
                  f"'tb_pubsub_case314_2', 'equal3');" \
                  f"insert into {self.tb_name3} values(3, '3');"
        pub_result = self.commsh_pub.execut_db_sql(
            sql_cmd, sql_type=self.user_param_pub)
        self.log.info(pub_result)
        self.assertEqual(pub_result.count(self.constant.INSERT_SUCCESS_MSG), 3,
                         '执行失败' + text)

        text = "step9:集群B订阅端查询数据是否更新 expect:表3数据不更新," \
               "其余数据更新"
        self.log.info(text)
        sql_select = f"select pg_sleep(5.5);select * from {self.tb_name1};" \
                     f"select * from {self.tb_name2};" \
                     f"select * from {self.tb_name3};"
        sub_result = self.commsh_sub.execut_db_sql(
            sql_select, self.user_param_sub, None, self.db_env_path_lite)
        self.log.info(sub_result)
        self.assertIn('3 | 3            | tb_pubsub_case314_1  | equal3',
                      sub_result, '执行失败' + text)
        self.assertIn('3 | 3            | tb_pubsub_case314_2  | equal3',
                      sub_result, '执行失败' + text)
        self.assertEqual(str(sub_result.splitlines()[-2]).strip(), '2 | 2',
                         '执行失败' + text)
        self.assertEqual(str(sub_result.splitlines()[-1]).strip(), '(1 row)',
                         '执行失败' + text)

        text = "step10:集群B发布端插入数据 expect:成功"
        self.log.info(text)
        sql_cmd = f"insert into {self.tb_name1} values(4, '4', " \
                  f"'tb_pubsub_case314_12', 'equal4');" \
                  f"insert into {self.tb_name2} values(4, '4', " \
                  f"'tb_pubsub_case314_22', 'equal4');" \
                  f"insert into {self.tb_name3} values(4, '4');"
        sub_result = self.commsh_sub.execut_db_sql(
            sql_cmd, self.user_param_sub, None, self.db_env_path_lite)
        self.log.info(sub_result)
        self.assertEqual(pub_result.count(self.constant.INSERT_SUCCESS_MSG), 3,
                         '执行失败' + text)

        text = "step11:集群A订阅端查询数据是否更新 expect:表2数据更新,其余" \
               "不更新"
        self.log.info(text)
        sql_select = f"select pg_sleep(5.5);select * from {self.tb_name1};" \
                     f"select * from {self.tb_name2};" \
                     f"select * from {self.tb_name3};"
        pub_result = self.commsh_pub.execut_db_sql(
            sql_select, sql_type=self.user_param_pub)
        self.log.info(pub_result)
        self.assertNotIn('4 | 4            | tb_pubsub_case314_22  | equal4',
                         pub_result, '执行失败' + text)
        self.assertEqual(str(pub_result.splitlines()[10]).strip(), '(3 rows)',
                         '执行失败' + text)
        self.assertIn('4 | 4            | tb_pubsub_case314_22 | equal4',
                      pub_result, '执行失败' + text)
        self.assertEqual(str(pub_result.splitlines()[-1]).strip(), '(2 rows)',
                         '执行失败' + text)
        self.assertEqual(str(pub_result.splitlines()[-2]).strip(), '3 | 3',
                         '执行失败' + text)
        self.assertEqual(str(pub_result.splitlines()[-3]).strip(), '1 | 1',
                         '执行失败' + text)

    def tearDown(self):
        text = "step12:清理环境 expect:成功"
        self.log.info(text)
        sql_cmd = f"drop subscription if exists {self.subname};"
        drop_sub_result = self.commsh_sub.execut_db_sql(
            sql_cmd, self.user_param_sub, None, self.db_env_path_lite)
        self.log.info(drop_sub_result)
        sql_cmd = f"drop publication if exists {self.pubname};"
        drop_pub_result = self.commsh_pub.execut_db_sql(
            sql_cmd, sql_type=self.user_param_pub)
        self.log.info(drop_pub_result)

        sql_cmd = f"drop subscription if exists {self.subname};"
        drop_pub_result1 = self.commsh_pub.execut_db_sql(
            sql_cmd, sql_type=self.user_param_pub)
        self.log.info(drop_pub_result1)
        sql_cmd = f"drop publication if exists {self.pubname};"
        drop_sub_result1 = self.commsh_sub.execut_db_sql(
            sql_cmd, self.user_param_sub, None, self.db_env_path_lite)
        self.log.info(drop_sub_result1)

        sql_cmd = f"drop table if exists {self.tb_name1},{self.tb_name2}," \
                  f"{self.tb_name3};"
        result_sub = self.commsh_sub.execut_db_sql(sql_cmd, self.user_param_sub, None,
                                                   self.db_env_path_lite)
        self.log.info(result_sub)
        result_pub = self.commsh_pub.execut_db_sql(
            sql_cmd, sql_type=self.user_param_pub)
        self.log.info(result_pub)
        cmd = f"mv " \
              f"{os.path.join(self.parent_path_pub, 'pg_hba.conf')} " \
              f"{os.path.join(macro.DB_INSTANCE_PATH, 'pg_hba.conf')}"
        self.log.info(cmd)
        sh_result = self.pri_userdb_pub.sh(cmd).result()
        self.log.info(sh_result)
        cmd = f"mv " \
              f"{os.path.join(self.parent_path_sub, 'pg_hba.conf')} " \
              f"{os.path.join(self.db_instance_path_lite, 'pg_hba.conf')}"
        self.log.info(cmd)
        sh_result1 = self.pri_userdb_sub.sh(cmd).result()
        self.log.info(sh_result1)
        result1 = self.commsh_pub.execute_gsguc(
            'set', self.constant.GSGUC_SUCCESS_MSG,
            f'wal_level={self.wal_level}')
        self.log.info(result1)
        result2 = self.commsh_sub.execute_gsguc(
            'set', self.constant.GSGUC_SUCCESS_MSG,
            f'wal_level={self.wal_level_sub}',
            single=True, dn_path=self.db_instance_path_lite,
            pghba_param=False, env_path=self.db_env_path_lite)
        result3 = self.commsh_pub.restart_db_cluster(True)
        result4 = self.commsh_sub.stop_db_instance(
            self.db_env_path_lite, self.db_instance_path_lite)
        result5 = self.commsh_sub.start_db_instance(
            env_path=self.db_env_path_lite,
            dn_path=self.db_instance_path_lite)
        self.assertTrue(result2, '执行失败' + text)
        self.assertTrue(result1, '执行失败' + text)
        self.assertTrue(result2, '执行失败' + text)
        self.assertTrue(result3, '执行失败' + text)
        self.assertTrue(result4, '执行失败' + text)
        self.assertTrue(result5, '执行失败' + text)
        self.assertEqual("", sh_result, '执行失败' + text)
        self.assertEqual("", sh_result1, '执行失败' + text)
        self.assertIn(self.constant.drop_pub_succ_msg, drop_pub_result,
                      '执行失败' + text)
        self.assertIn(self.constant.drop_sub_succ_msg, drop_sub_result,
                      '执行失败' + text)
        self.assertIn(self.constant.drop_pub_succ_msg, drop_sub_result1,
                      '执行失败' + text)
        self.assertIn(self.constant.drop_sub_succ_msg, drop_pub_result1,
                      '执行失败' + text)
        self.assertIn(self.constant.DROP_TABLE_SUCCESS, result_sub, '执行失败'
                      + text)
        self.assertIn(self.constant.DROP_TABLE_SUCCESS, result_pub, '执行失败'
                      + text)
        self.log.info(f"-----{os.path.basename(__file__)} end-----")
